package edu.vt.ece.rt.quorum.quorum;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import org.jgroups.Address;
import org.jgroups.Message;

import edu.vt.ece.rt.benchmark.BenchmarkThread;
import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.quorum.Cluster;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.MessageHandler;
import edu.vt.ece.rt.quorum.messsages.types.CommitRequest;
import edu.vt.ece.rt.quorum.messsages.types.JoinCluster;
import edu.vt.ece.rt.quorum.messsages.types.PublishObject;
import edu.vt.ece.rt.quorum.messsages.types.ReadRequest;
import edu.vt.ece.rt.quorum.tree.Node;
import edu.vt.ece.rt.quorum.tree.Tree;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.quorum.util.ReplyListStore;
import edu.vt.ece.rt.quorum.util.ReplyObjectStore;
import edu.vt.ece.rt.util.Logger.Logger;

public class WriteQuorum {
	/*
	 * Map for transaction ID to object list, utilised while request commit, object add to list
	 * only if write request is made
	 */	
	ConcurrentHashMap<Integer, List<QObject>> objMap;
	
	ReadQuorum rQ;			//A List of nodes representing readers
	List<Node> writers;		//A List of nodes representing write 
	Cluster writeCluster;

	private boolean partOfTree = false;

	// This will create WQ on given tree, the tree can have failure/down nodes
	public void createWriteQuorum(Tree t) {
		
		ArrayList<Node> wq = new ArrayList<Node>();
		if (createWriteQuorumList(t, wq)) {
			for (int i = 0; i < wq.size(); i++) {
				Logger.debug("Node in WQ is "+wq.get(i).address);
			}
			
			writers = wq;
			
			// Check if part of tree
			Address adr = ClustersManager.getCluster(Names.base).getMyAddress();
			for(Node n: writers){
				if(adr.equals(n.address)){
					partOfTree = true;
					break;
				}					
			}
		}
		
	}
	
	
	public boolean createWriteQuorumList(Tree t, ArrayList<Node> wq) {
		
		if (t.root.isRunning()) {
			Logger.debug("Node added "+ t.root.address);
			wq.add(t.root);
		} else {
			Logger.debug("Node not running "+ t.root.val);
			return false;
		}
		
		// for all children {
		//		if (createWriteQuorumList(c)) { num++}
		//		if (num > majority) great!
		// }
		// if (num < majority) false
		
		int numWQ = 0;
		int majority = t.degree/2 + 1;
		
		if (t.childList == null || t.childList.size() == 0)
			return true;
		
		for (int i = 0; i < t.childList.size(); i++) {
			
			ArrayList<Node> tmp =  new ArrayList<Node>();
			if (createWriteQuorumList(t.childList.get(i), tmp)) {
				wq.addAll(tmp);
				numWQ++;
			}
			if (numWQ >= majority) return true;
		}
		
		return false;
	}

	/*
	 * Returns True if it is able to commit successfully. 
	 */	
	public Boolean requestCommit(Long tnx, int tries) throws Exception{
		ReplyListStore rl = new ReplyListStore(getWriteReplyCount());
		Logger.debug("Sending the Commit Request for "+tnx+" will wait for "+getWriteReplyCount());
		String messageId =  Long.toString(tnx) + "-" + Integer.toString(tries);
		Logger.debug("sending message Id "+messageId);
		
		synchronized (rl) {								//Get in monitor
			MessageHandler.commitRequestQueue.put(tnx, rl);
			writeCluster.send(new Message(null, new CommitRequest(ContextDelegator.getContext(tnx), ClustersManager.baseClusterId,ContextDelegator.getContext(tnx).getParentContextId(), messageId)));	//Send request now
			//Wait for all the requests, will notified by ReadResponse if last message
			rl.wait(ClustersManager.ClusterWaitTime);
//			rl.wait();
			MessageHandler.commitRequestQueue.remove(tnx);			//Remove the value from HashMap
		}
		
		if (Thread.currentThread() instanceof BenchmarkThread)
			((BenchmarkThread)Thread.currentThread()).incrementMessage(getWriteReplyCount());
		else 
			Resource.incrementMessages(getWriteReplyCount());


		if(rl.requestFail) {
//			Logger.debug("Commit fail due to Quorum failure " +tnx);
//			throw new QuroumFailureException();
			return null;
		}
		
		if(ContextDelegator.getContext(tnx).isAborted()){
			Logger.debug("XXXX Commit Fail XXXX " +tnx);
//			System.out.println("XXXX Commit Fail XXXX " +tnx);
			return false;
		}
		
		else if(ContextDelegator.getContext(tnx).isCommitted()){
			if (ContextDelegator.getContext(tnx).isChild()) {
				// merge the read/write set with parent
				ContextDelegator.getContext(tnx).mergeParent();
			}
//			System.out.println("**** Commit Success ***** "+tnx);
			Logger.debug("**** Commit Success ***** "+tnx);
			return true;
		}
		return false;
	}
	

	/**
	 * Provides the number of message required to be waited
	 * @return
	 */
	private int getWriteReplyCount() {
//		if(partOfTree)
//			return writers.size() - 1 ;
//		else{
//			return writers.size();
//		}
		//Should be writers.size() as nothing done using local access
		return writers.size();
	}


	/**
	 * @return
	 */
	public static WriteQuorum getQuorum() {
		WriteQuorum wrq = new WriteQuorum();
		wrq.createWriteQuorum(ClustersManager.getCluster(Names.base).getTree());
		if(wrq.writers == null){
			return null;
		}
		
		//Join your self
		String groupName = ClustersManager.baseClusterId + Names.writeQuorum;
		Cluster c = new Cluster(groupName);
		try {
			c.init();
		} catch (Exception e) {
			e.printStackTrace();
		}
		//Register the cluster
		ClustersManager.registerCluster(groupName, c);
		
		//Invite others
		for(Node n: wrq.writers){
			//By default auto-receive off , so will not receive join request if part of tree quorum
			try {
				ClustersManager.getCluster(Names.base).send(new Message(n.address, new JoinCluster(groupName)));
				try {
					Thread.sleep((20000L));
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		//Wait all to join
		c.waitForCluster(new Integer(wrq.getWaitCount()));
        System.out.println("Wait over for "+ groupName +" is "+ c.currentView.size());

		Logger.debug("WriteQuorum Created");
		wrq.writeCluster = c;
		
		if(wrq.partOfTree)
			wrq.writeCluster.setAutoReceive(true);
		
		return wrq;
	}


	/**
	 * Gives the count of nodes for which cluster need to wait to join
	 * @return
	 */
	private int getWaitCount() {
		if(partOfTree ){
			return writers.size();
		}else{
			return writers.size()+1;
		}
	}


	/**
	 * @return
	 */
	public Cluster getWriters() {
		return writeCluster;
	}
	
	/**Adds the object to quorum, if uninitiated context
	 * It publishes the object too.
	 * @return
	 */
	public void addToQuorum(QObject obj){
		Context c = ContextDelegator.getContext();
		c.addToCreateSet(obj);		
		
//		if(c.getContextId() == null){
			try {
				publishObjects(c);
			} catch (Exception e) {
				e.printStackTrace();
//			}
		}
	}
	
	/**Adds multiple objects to quorum, if uninitiated context
	 * It publishes the object too.
	 * @return
	 */
	public void addToQuorum(List<QObject> l){
		Context c = ContextDelegator.getContext();
		
		for(QObject obj: l)
			c.addToCreateSet(obj);		
			
		if(c.getContextId() == null){
			try {
				publishObjects(c);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * removes the object to quorum
	 * @return
	 */
	public void removeFromQuorum(QObject obj){
		Context c = ContextDelegator.getContext();
		c.addToDeleteSet(obj);	
		
		//Should not be required to call uninitiated context to remove
		//Still implemented 
		if(c.getContextId() == null){
			try {
				publishObjects(c);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * removes the list of object from quorum
	 * @return
	 */
	public void removeFromQuorum(List<QObject> l){
		Context c = ContextDelegator.getContext();
		
		for(QObject obj: l)
			c.addToDeleteSet(obj);	
		
		//Should not be required to call uninitiated context to remove
		//Still implemented 
		if(c.getContextId() == null){
			try {
				publishObjects(c);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	/**Publish the object in the quorum create
	 * @return
	 * @throws Exception 
	 */
	public void publishObjects(Context c) throws Exception{
		if(c != null) {
			ClustersManager.getCluster(Names.base).send(new Message(null, new PublishObject(c)));// broadcast
			System.out.println("Broadcasting");
		}
	}	
	
	/**
	 * @param id
	 * @param tnx
	 * @param b
	 * @return
	 * @throws Exception 
	 */
	public QObject write(String id, Long tnx, boolean b, int tries) throws Exception {
		// Create a waiting point
		ReplyObjectStore ws = new ReplyObjectStore(getWriteReplyCount());
		Logger.debug("Write will wait for "+ getWriteReplyCount());
		Context c = ContextDelegator.getContext(tnx);
		Long txnIdParent = c.getParentContextId();
		String messageId = Long.toString(c.getContextId()) + "-" + Integer.toString(tries);
		Logger.debug("sending message Id "+messageId);
		
		synchronized (ws) {								//Get in monitor
			String key = tnx + "-" + id;				//TODO: Move to a function
			MessageHandler.wobjectRequestQueue.put(key, ws);
			writeCluster.send(new Message(null, new ReadRequest(id, tnx, txnIdParent, b, ClustersManager.baseClusterId, messageId)));	//Send request now
			//Wait for all the requests, will notified by ReadResponse if last message
			ws.wait(ClustersManager.ClusterWaitTime);											
			MessageHandler.wobjectRequestQueue.remove(key);			//Remove the value from HashMap
		}
		
		
		if (Thread.currentThread() instanceof BenchmarkThread)
			((BenchmarkThread)Thread.currentThread()).incrementMessage(getWriteReplyCount());
		else 
			Resource.incrementMessages(getWriteReplyCount());


		//Check request failed
		if(ws.requestFail) {
//			System.out.println("Quorum for write exception for node "+ ClustersManager.baseClusterId+" !!!!");
//			throw new QuroumFailureException();
			return null;
		}
		
		//else go ahead
		QObject obj = ws.findLatest();
		if(obj == null)
			return null;
		
		//Add Object to context readWriteSet
		if(b){
			c.addToReadSet(obj);
			Logger.debug("Added "+obj.getId()+" to read set of "+c.getContextId());
			return obj;
		} else{
			c.addToWriteSet(obj);				//Add to write set of context
			Logger.debug("Added "+obj.getId()+" to write set of "+c.getContextId());
			return obj;
		}
	}

	public static void main(String[] args) {
		ArrayList<Node> l= new ArrayList<Node>();
		for(int i=0;i<13;i++){
			l.add(new Node(i));
		}
		
		Tree t = Tree.createTree(l, 3);
		
		t.print();
		
		WriteQuorum wq = new WriteQuorum();
		wq.createWriteQuorum(t);
		
	}
}